package com.iipaas.twheel;

import cn.hutool.core.thread.ExecutorBuilder;

import java.util.ArrayList;
import java.util.List;
import java.util.concurrent.ExecutorService;
import java.util.concurrent.atomic.AtomicBoolean;
import java.util.concurrent.locks.ReadWriteLock;
import java.util.concurrent.locks.ReentrantReadWriteLock;

/**
 * 时间轮
 * @author chin
 */
public class TimingWheel<T> {
    private final long maxTime;
    /**
     * 每个槽表示的时间（毫秒）
     */
    private final long ticksPerWheel;

    /**
     * 槽数
     */
    private final int size;

    /**
     * 时间轮的时间跨度
     */
    private final long interval;

    /**
     * 槽
     */
    private final List<Slot<T>> slots;
    /**
     * 启动标识
     */
    private final AtomicBoolean running = new AtomicBoolean(false);

    /**
     * 锁
     */
    private final ReadWriteLock lock = new ReentrantReadWriteLock();

    private final List<ExpirationListener<T>> listeners;

    /**
     * 当前槽
     */
    private int current;

    /**
     * 是否是分层
     */
    private final boolean hierarchical;

    /**
     * 当前最大运行的时间
     */
    private final int currentLevel;
    private TimingWheel<T> next;
    private final TimingWheel<T> prev;


    private final ExecutorService pool;
    private final ExecutorService taskPool;

    private TimingWheel(long ticksPerWheel, int size, long maxTime,
                        List<ExpirationListener<T>> listeners,
                        int taskCorePoolSize, int taskMaxPoolSize,
                        boolean hierarchical, TimingWheel<T> prev, int level) {
        this.ticksPerWheel = ticksPerWheel;
        this.size = size;
        this.maxTime = maxTime;
        this.listeners = listeners;
        this.interval = ticksPerWheel * size;
        this.slots = new ArrayList<>();
        for (int i = 0; i < this.size; i ++) {
            slots.add(new Slot<>());
        }
        this.pool = level == 1 ? ExecutorBuilder.create().setCorePoolSize(1).setMaxPoolSize(1).setKeepAliveTime(0).build() : null;
        this.taskPool = level == 1 ? ExecutorBuilder.create().setCorePoolSize(taskCorePoolSize).setMaxPoolSize(taskMaxPoolSize).build() : null;
        this.hierarchical = hierarchical;
        this.currentLevel = level;
        this.prev = prev;
    }

    public void add(T t, long timeout) {
        synchronized (slots) {
            if (hierarchical) {
                addHierarchical(t, timeout);
            } else {
                addSingle(t, timeout);
            }
        }
        if (currentLevel == 1) {
            start();
        }
    }

    private void addSingle(T t, long timeout) {
        if (timeout >= interval) {
            return;
        }
        Slot<T> slot = slots.get((int) (((timeout / ticksPerWheel) + current) % size));
        slot.add(t, timeout);
    }

    private void addHierarchical(T t, long timeout) {
        // 判断是否是当前层
        if (timeout < interval) {
            addSingle(t, timeout);
        } else {
            if(this.next == null) {
                this.next = new TimingWheel<>(interval, size, maxTime, null,
                        0,  0, hierarchical, this, currentLevel+1);
            }
            this.next.add(t, timeout - (size - current) * ticksPerWheel);
        }
    }

    public boolean isRunning() {
        return this.running.get();
    }

    public void start() {
        if (!this.running.get() && this.running.compareAndSet(false, true)) {
            pool.execute(new TickWorker<>(this));
        }
    }

    public void stop() {
        if (!pool.isShutdown()) {
            pool.shutdown();
        }
        if (!taskPool.isShutdown()) {
            taskPool.shutdown();
        }
        if (this.running.compareAndSet(true, false)) {
            pool.shutdownNow();
            taskPool.shutdownNow();
        }
    }

    public void writeLock() {
        lock.writeLock().lock();
    }

    public void writeUnLock() {
        lock.writeLock().unlock();
    }

    public void tick() {
        Slot<T> slot = slots.get(current);
        if (!hierarchical || currentLevel == 1) {
            if (listeners != null) {
                slot.forEach(t -> listeners.forEach(listener -> taskPool.execute(listener.expired(t.getData()))));
            }
        }
        if (currentLevel > 1) {
            // 推送数据给上一层
            slot.forEach(task -> this.prev.add(task.getData(), task.timeout()));
        }
        slot.clear();
        current ++;
        if (current == size) {
            current = 0;
            if (this.next != null) {
                this.next.tick();
            }
        }
    }

    public long ticksPerWheel() {
        return this.ticksPerWheel;
    }

    public static <T> Builder<T> builder() {
        return new Builder<>();
    }

    public static class Builder<T> {
        private long ticksPerWheel = 1;
        private int size = 20;
        private long maxTime = Long.MAX_VALUE;
        private List<ExpirationListener<T>> listeners;
        private int taskCorePoolSize = 5;
        private int taskMaxPoolSize = 10;
        private boolean hierarchical = true;

        public Builder<T> ticksPerWheel(long ticksPerWheel) {
            this.ticksPerWheel = ticksPerWheel <=0 ? 1 : ticksPerWheel;
            return this;
        }

        public Builder<T> size(int size) {
            this.size = size <=0 ? 20 : size;
            return this;
        }

        public Builder<T> maxTime(long maxTime) {
            this.maxTime = maxTime <=0 ? 1 : maxTime;
            return this;
        }

        public Builder<T> listeners(List<ExpirationListener<T>> listeners) {
            this.listeners = listeners;
            return this;
        }

        public Builder<T> taskCorePoolSize(int taskCorePoolSize) {
            this.taskCorePoolSize = taskCorePoolSize <=0 ? 5 : taskCorePoolSize;
            return this;
        }

        public Builder<T> taskMaxPoolSize(int taskMaxPoolSize) {
            this.taskMaxPoolSize = taskMaxPoolSize <=0 ? 10 : taskMaxPoolSize;
            return this;
        }

        public Builder<T> hierarchical(boolean hierarchical) {
            this.hierarchical = hierarchical;
            return this;
        }

        public TimingWheel<T> build() {
            return new TimingWheel<>(ticksPerWheel, size, maxTime,
                    listeners, taskCorePoolSize, taskMaxPoolSize,
                    hierarchical, null, 1);
        }

    }
}
